package com.heima.stream.listener;

import com.heima.stream.processor.WordProcessor;
import com.sun.org.apache.regexp.internal.RE;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import sun.rmi.runtime.Log;

import javax.xml.crypto.dsig.keyinfo.KeyValue;
import java.security.Key;
import java.time.Duration;
import java.time.temporal.ValueRange;
import java.util.Arrays;
//开发流式计算应用-SpringCloudStream 1.新建处理器接口 @Input数据来源 @Output计算结果 2.编写流式计算业务类 通过@EnableBind 绑定处理器接口
//@StreamListener("源头数据话题") 编写kafka内部对于数据计算 分析逻辑  通过@SendTo("结果话题")将计算结果发送到结果话题也叫目标话题
/**
 * 流式应用监听器：绑定处理器接口；流式计算业务逻辑
 * @author: itheima
 * @create: 2022-07-13 12:01
 */
@Component
@EnableBinding(WordProcessor.class) //绑定处理器接口
public class WordListener {


    /**
     * 流式计算业务逻辑：统计最近一段时间内，评论内容相同次数
     *
     * @param input 监听到原始话题中的数据 数据流 得到结果
     * @return 将统计个数发到目标话题中
     */
    @StreamListener("words_1_input") //数据来源
    @SendTo("counts_1_output")//计算结果
    public KStream<String, String> processor(KStream<String, String> input){
        //1.从数据流对象中获取记录  将val值进行采用空格分隔
       input.flatMapValues(new ValueMapper<String,Iterable<String>>(){

           @Override
           public  Iterable<String> apply(String value){
               log.info("监听到原始流中数据：{}",value);
               String[] s = value.split(" ");
               return  Arrays.asList(s);
           }
        });
        //2.对以上处理后数据进行分组,根据值进行分组：将值相同放一组
        KGroupedStream<String, String> groupedStream = flatMapValues.groupBy(new KeyValueMapper<String, String, String>() {
            @Override
            public String apply(String key, String value) {
                log.info("分组字段: {}" ,value);
                //根据值进行分组
                return value;
            }
        });

        //3.对以上处理后数据进行设置时间窗口 ，统计流中最近一段时间数据   10s
        TimeWindowedKStream<String, String> timeWindowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(20)));

        //4.对以上处理后数据进行统计 注意 时间窗口+分组 Key类型：是 Windowed<T> 类型 T是分组后内容
        KTable<Windowed<String>, Long> count = timeWindowedKStream.count();
        //5.对流中数据键值变更类型 KeyValueMapper<键类型，值类型，处理后新键值对象<键类型,值类型>> 因为返回的类型必须和yml配置文件中配的序列化器类型一致
        KStream<String, String> KStream= count.toStream().map(new KeyValueMapper<Windowed<String>,Long, KeyValue<String,String>>(){

        /**
         * 处理后键值对象
         * @param key
         * @param value
         * @return
         */
          @Override
            public KeyValue<String,String> apply(Windowed<String> key, Long value){
              log.info("处理流中数据类型，处理key：Windowed<String>转为String ，处理Val将Long转为String");
              return  new KeyValue<>(key.key(), value.toString()); //返回给第三个即处理后新键值对象<键类型,值类型>>  KeyValue<String,String>了
        }

    });
    return  KStream;
}
}